package com.telra.app.server.esper;

/**
 * Receives JMS order updates from the Listener class via EPServiceProvider.
 * CEP engine processes different expressions and fowards aggregate updates from
 * CEP to AggregateListener.
 *
 */
import java.text.DecimalFormat;

import com.telra.app.server.data.OrderEvent;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;

import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.UpdateListener;

public class EsperService {

    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(EsperService.class);
    private final EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider();
    private final static DecimalFormat fmt = new DecimalFormat("#,##0.00");
    private AggregateListener agglistener;

    public EsperService() {

        agglistener = new AggregateListener();

        String expression = "select count(*) as count, avg(price), sum(price), max(price), category " +
                "from com.telra.app.server.data.OrderEvent " +
                "group by category";
        EPStatement statement = epService.getEPAdministrator().createEPL(expression);
        statement.addListener(new OrderUpdateListener());

        expression = "select count(*) as count, avg(price), max(price) from com.telra.app.server.data.OrderEvent.win:time(30 sec)";
        statement = epService.getEPAdministrator().createEPL(expression);
        statement.addListener(new OrderWindowUpdateListener());
    }

    private class OrderWindowUpdateListener implements UpdateListener {

        public void update(EventBean[] newEvents, EventBean[] oldEvents) {
            EventBean event = newEvents[0];
            log.debug("30 second: count="+ event.get("count") + ", avg=" +
                    event.get("avg(price)") + ", max=" + event.get("max(price)") );
        }
    }

    private class OrderUpdateListener implements UpdateListener {

        public void update(EventBean[] newEvents, EventBean[] oldEvents) {

            for (EventBean event : newEvents) {
                if (log.isDebugEnabled())   {
                    log.debug(new StringBuilder().append("count=").append(event.get("count")).
                            append(", category=").append(event.get("category")).append(", sum=").
                            append(fmt.format(event.get("sum(price)"))).append(", avg=").
                            append(fmt.format(event.get("avg(price)"))).append(", max=").
                            append(fmt.format(event.get("max(price)"))).toString());
                }
                agglistener.update(
                        (OrderEvent.Category) event.get("category"),
                        ((Long) event.get("count")).intValue(),
                        (Double) event.get("avg(price)"),
                        (Double) event.get("sum(price)"),
                        (Double) event.get("max(price)")
                );

            }

        }

    }

    public EPServiceProvider getEPServiceProvider() {
        return epService;
    }

}
